use crate::common::{config::ServerConfig, errors::Result, messages::*};
use crate::server::registry::NODE_REGISTRY;
use crate::common::storage::Storage;
use crate::server::pulsar_bus::PulsarBus;
use crate::common::proto::oc_cluster;
use crate::common::metrics::METRICS;
use axum::{
    routing::{get, post, put, delete},
    http::{StatusCode, header},
    Json, Router,
    extract::State,
    response::IntoResponse,
};
use std::sync::Arc;
use tracing::{info, error, debug};
use serde::{Deserialize, Serialize};
use crate::common::storage::{Group, GroupSelector, CommandStatus};

#[derive(Clone)]
pub struct AppState {
    pub config: ServerConfig,
    pub storage: std::sync::Arc<dyn Storage>,
    pub bus: std::sync::Arc<PulsarBus>,
}

pub async fn start_http_server(config: ServerConfig, storage: std::sync::Arc<dyn Storage>, bus: std::sync::Arc<PulsarBus>) -> Result<()> {
    let state = AppState { config, storage, bus };
    
    let app = Router::new()
        .route("/health", get(health_check))
        .route("/metrics", get(metrics))
        .route("/api/v1/nodes", get(list_nodes))
        .route("/api/v1/nodes/:id", get(get_node))
        .route("/api/v1/groups", get(list_groups).post(create_group))
        .route("/api/v1/groups/:name/nodes", get(list_group_nodes))
        .route("/api/v1/groups/:id", put(update_group).delete(delete_group))
        .route("/api/v1/nodes/register", post(register_node))
        .route("/api/v1/commands/:id", get(get_command))
        .route("/api/v1/commands/target/:target", get(get_commands_by_target))
        .route("/api/v1/commands", post(execute_command))
        .route("/api/v1/files", post(transfer_file))
        .with_state(Arc::new(state));
    
    // 正确绑定监听地址：使用 server.host 与 server.port 或独立 http 端口
    let addr = format!("{}:{}", config.server.host, config.server.http_port);
    info!("HTTP API server starting on {}", addr);
    
    axum::Server::bind(&addr.parse()?)
        .serve(app.into_make_service())
        .await?;
    
    Ok(())
}

async fn health_check() -> StatusCode {
    StatusCode::OK
}

async fn metrics() -> impl IntoResponse {
    let body = METRICS.encode_prometheus();
    (
        StatusCode::OK,
        [(header::CONTENT_TYPE, "text/plain; version=0.0.4; charset=utf-8")],
        body,
    )
}

async fn list_nodes(
    State(state): State<Arc<AppState>>,
) -> Json<serde_json::Value> {
    let nodes = state.storage.get_all_nodes().await.unwrap_or_default();
    Json(serde_json::json!({
        "nodes": nodes,
        "total": nodes.len()
    }))
}

async fn get_node(
    State(state): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
) -> Json<serde_json::Value> {
    let node = state.storage.get_node(&id).await.unwrap_or(None);
    Json(serde_json::json!({
        "node": node
    }))
}

async fn list_groups() -> Json<serde_json::Value> {
    // 先提供示例分组（后续可接入DB/Redis）
    Json(serde_json::json!({
        "groups": [
            {"name": "all", "selector": {"type": "AllNodes"}},
        ]
    }))
}

async fn list_group_nodes(axum::extract::Path(name): axum::extract::Path<String>) -> Json<serde_json::Value> {
    let nodes = if name == "all" {
        NODE_REGISTRY.list()
    } else {
        // 标签、IP等选择器匹配可在此实现
        vec![]
    };
    Json(serde_json::json!({
        "group": name,
        "nodes": nodes
    }))
}

#[derive(Deserialize)]
struct RegisterNodeRequest {
    id: String,
    hostname: String,
    ip_address: String,
    tags: Vec<String>,
}

async fn register_node(
    State(state): State<Arc<AppState>>,
    Json(payload): Json<RegisterNodeRequest>,
) -> impl IntoResponse {
    let node = crate::common::storage::NodeInfo {
        id: payload.id,
        hostname: payload.hostname,
        ip_address: payload.ip_address,
        tags: payload.tags,
        status: crate::common::storage::NodeStatus::Online,
        last_heartbeat: chrono::Utc::now(),
        capabilities: vec![],
        metadata: Default::default(),
        group_ids: vec![],
    };
    match state.storage.register_node(&node).await {
        Ok(_) => (StatusCode::OK, Json(serde_json::json!({"status":"ok"}))).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": e.to_string()}))).into_response(),
    }
}

#[derive(Deserialize)]
struct CreateGroupRequest {
    id: String,
    name: String,
    description: Option<String>,
    selector: Option<serde_json::Value>,
    nodes: Option<Vec<String>>,
}

async fn create_group(
    State(state): State<Arc<AppState>>,
    Json(payload): Json<CreateGroupRequest>,
) -> impl IntoResponse {
    let selector = payload.selector
        .and_then(|v| serde_json::from_value::<GroupSelector>(v).ok())
        .unwrap_or(GroupSelector::AllNodes);
    let group = Group {
        id: payload.id,
        name: payload.name,
        description: payload.description.unwrap_or_default(),
        selector,
        nodes: payload.nodes.unwrap_or_default(),
        created_at: chrono::Utc::now(),
        updated_at: chrono::Utc::now(),
    };
    match state.storage.create_group(&group).await {
        Ok(_) => (StatusCode::OK, Json(serde_json::json!({"status":"ok"}))).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": e.to_string()}))).into_response(),
    }
}

#[derive(Deserialize)]
struct UpdateGroupRequest {
    name: Option<String>,
    description: Option<String>,
    selector: Option<serde_json::Value>,
    nodes: Option<Vec<String>>,
}

async fn update_group(
    State(state): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
    Json(payload): Json<UpdateGroupRequest>,
) -> impl IntoResponse {
    match state.storage.get_group(&id).await {
        Ok(Some(mut group)) => {
            if let Some(name) = payload.name { group.name = name; }
            if let Some(desc) = payload.description { group.description = desc; }
            if let Some(sel) = payload.selector {
                if let Ok(parsed) = serde_json::from_value::<GroupSelector>(sel) { group.selector = parsed; }
            }
            if let Some(nodes) = payload.nodes { group.nodes = nodes; }
            group.updated_at = chrono::Utc::now();
            match state.storage.update_group(&group).await {
                Ok(_) => (StatusCode::OK, Json(serde_json::json!({"status":"ok"}))).into_response(),
                Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": e.to_string()}))).into_response(),
            }
        }
        Ok(None) => (StatusCode::NOT_FOUND, Json(serde_json::json!({"error":"group not found"}))).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": e.to_string()}))).into_response(),
    }
}

async fn delete_group(
    State(state): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
) -> impl IntoResponse {
    match state.storage.delete_group(&id).await {
        Ok(_) => (StatusCode::OK, Json(serde_json::json!({"status":"ok"}))).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": e.to_string()}))).into_response(),
    }
}

async fn get_command(
    State(state): State<Arc<AppState>>,
    axum::extract::Path(id): axum::extract::Path<String>,
) -> impl IntoResponse {
    match state.storage.get_command(&id).await {
        Ok(Some(cmd)) => (StatusCode::OK, Json(serde_json::json!({"command": cmd}))).into_response(),
        Ok(None) => (StatusCode::NOT_FOUND, Json(serde_json::json!({"error":"not found"}))).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": e.to_string()}))).into_response(),
    }
}

async fn get_commands_by_target(
    State(state): State<Arc<AppState>>,
    axum::extract::Path(target): axum::extract::Path<String>,
) -> impl IntoResponse {
    match state.storage.get_commands_by_target(&target).await {
        Ok(cmds) => (StatusCode::OK, Json(serde_json::json!({"commands": cmds}))).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": e.to_string()}))).into_response(),
    }
}

#[derive(Deserialize)]
struct CommandRequest {
    command: String,
    target_groups: Vec<String>,
    timeout: Option<u64>,
}

async fn execute_command(
    State(state): State<Arc<AppState>>,
    Json(payload): Json<CommandRequest>,
) -> Json<serde_json::Value> {
    let command_id = uuid::Uuid::new_v4();
    
    // 基于组名解析节点列表（all / tag:foo / ip:10.0.）
    let mut targets = Vec::new();
    for g in &payload.target_groups {
        if g == "all" {
            targets.extend(NODE_REGISTRY.list());
        } else if let Some(tag) = g.strip_prefix("tag:") {
            targets.extend(NODE_REGISTRY.select_by_tags(&vec![tag.to_string()]));
        } else if let Some(prefix) = g.strip_prefix("ip:") {
            targets.extend(NODE_REGISTRY.select_by_ip_prefix(prefix));
        }
    }
    // 去重
    targets.sort_by(|a,b| a.id.cmp(&b.id));
    targets.dedup_by(|a,b| a.id == b.id);

    info!("Executing command: {} on targets: {} nodes", payload.command, targets.len());

    // Pulsar 生产者
    let topic = format!("{}/commands", state.config.pulsar.topic_prefix);
    let producer = match state.bus.create_producer(&topic).await {
        Ok(p) => p,
        Err(e) => {
            error!("pulsar producer error: {}", e);
            return Json(serde_json::json!({"error": e.to_string()}));
        }
    };
    // 构造 Protobuf Command
    use prost::Message;
    let cmd = oc_cluster::Command {
        id: command_id.to_string(),
        command_type: oc_cluster::CommandType::Shell as i32,
        target: serde_json::to_string(&targets).unwrap_or_default(),
        payload: payload.command.as_bytes().to_vec(),
        timeout: payload.timeout.unwrap_or(60) as i32,
        r#async: true,
    };
    let mut buf = Vec::new();
    cmd.encode(&mut buf).unwrap();
    let _ = producer.send(buf).await;
    Json(serde_json::json!({
        "command_id": command_id.to_string(),
        "status": "queued",
        "message": "Command queued for execution",
        "target_nodes": targets
    }))
}

#[derive(Deserialize)]
struct FileTransferRequest {
    filename: String,
    target_groups: Vec<String>,
}

async fn transfer_file(
    State(state): State<Arc<AppState>>,
    Json(payload): Json<FileTransferRequest>,
) -> Json<serde_json::Value> {
    let file_id = uuid::Uuid::new_v4();
    
    let mut targets = Vec::new();
    for g in &payload.target_groups {
        if g == "all" {
            targets.extend(NODE_REGISTRY.list());
        } else if let Some(tag) = g.strip_prefix("tag:") {
            targets.extend(NODE_REGISTRY.select_by_tags(&vec![tag.to_string()]));
        } else if let Some(prefix) = g.strip_prefix("ip:") {
            targets.extend(NODE_REGISTRY.select_by_ip_prefix(prefix));
        }
    }
    targets.sort_by(|a,b| a.id.cmp(&b.id));
    targets.dedup_by(|a,b| a.id == b.id);

    info!("Transferring file: {} to {} nodes", payload.filename, targets.len());

    // Pulsar 文件分发任务
    let topic = format!("{}/files", state.config.pulsar.topic_prefix);
    let producer = match state.bus.create_producer(&topic).await {
        Ok(p) => p,
        Err(e) => {
            error!("pulsar producer error: {}", e);
            return Json(serde_json::json!({"error": e.to_string()}));
        }
    };
    use prost::Message;
    let ft = oc_cluster::FileTransfer {
        id: file_id.to_string(),
        filename: payload.filename.clone(),
        file_size: 0,
        checksum: String::new(),
        chunk_size: 0,
        total_chunks: 0,
        transfer_type: oc_cluster::FileTransferType::Upload as i32,
    };
    let mut buf = Vec::new();
    ft.encode(&mut buf).unwrap();
    let _ = producer.send(buf).await;
    Json(serde_json::json!({
        "file_id": file_id.to_string(),
        "status": "queued",
        "message": "File transfer queued",
        "target_nodes": targets
    }))
} 